package com.lry.persist;

import com.lry.MixAll;
import com.lry.ServiceThread;
import com.lry.message.MessageQueue;
import com.lry.serialize.KryoSerializer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


@Slf4j
public class TopicPersist extends ServiceThread implements Persist{

    private Map<String, List<MessageQueue>> topicQueue;

    private String topicStorePath;

    public Map<String, List<MessageQueue>> getTopicQueue() {
        return topicQueue;
    }

    public TopicPersist(String topicStorePath){
        this.topicQueue = new ConcurrentHashMap<>();
        this.topicStorePath = topicStorePath;
    }

    @Override
    public String getServiceName() {
        return TopicPersist.class.getSimpleName();
    }

    @Override
    public long getJointime() {
        return 1000 * 60 * 5;
    }

    @Override
    public void run() {
        TopicPersist.log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                this.waitForRunning(1000);
                this.doPersist();
            } catch (Exception e) {
                TopicPersist.log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            TopicPersist.log.warn(this.getServiceName() + " Exception, ", e);
        }

        this.doPersist();
        TopicPersist.log.info(this.getServiceName() + " service end");
    }

    @SneakyThrows
    @Override
    public void doPersist() {
        if(topicQueue.isEmpty()){
            return;
        }
        byte[] bytes = KryoSerializer.serialize(topicQueue);
        MixAll.bytes2File(bytes,topicStorePath);
    }

    @Override
    public void load() {
        byte[] bs = null;
        try {
            bs = MixAll.file2Bytes(topicStorePath);
            if(null!=bs){
                this.topicQueue =KryoSerializer.deserialize(bs);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Map<String,List<MessageQueue>> map = new ConcurrentHashMap<>();
        List<MessageQueue> list = new ArrayList<>();
        MessageQueue messageQueue = new MessageQueue("topic", "dsaf", 21);
        list.add(messageQueue);
        map.put("topic", list);

        byte[] bytes = KryoSerializer.serialize(map);

        Map<String,List<MessageQueue>> map1 =KryoSerializer.deserialize(bytes);
        System.out.println(map1);
    }

}
